package com.ferry.lock.engine.zookeeper.menagerie;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by daiyong
 * 原始数据
 */
public class ZkPrimitive {

	protected static final byte[] emptyNode = new byte[]{};

	protected final ZKSessionManager zkSessionManager;

	protected final String baseNode;

	protected final List<ACL> privileges;

	protected final Lock localLock;

	protected final Condition condition;

	protected volatile boolean broken = false;

	protected final ConnectionListener connectionListener = new PrimitiveConnectionListener(this);

	protected final Watcher signalWatcher;

	protected ZkPrimitive(String baseNode, ZKSessionManager zkSessionManager, List<ACL> privileges) {
		if(baseNode == null)
			throw new NullPointerException("No base node specified!");
		this.baseNode = baseNode;
		this.zkSessionManager = zkSessionManager;
		this.privileges = privileges;
		this.localLock = new ReentrantLock(true);
		condition = this.localLock.newCondition();
		signalWatcher = new SignallingWatcher(this);
		ensureNodeExists();
	}

	protected final void ensureNodeExists(){
		try {
			ZooKeeper zooKeeper = zkSessionManager.getZooKeeper();

			Stat stat = zooKeeper.exists(baseNode, false);
			if(stat==null){
				zooKeeper.create(baseNode,emptyNode,privileges, CreateMode.PERSISTENT);
			}

		} catch (KeeperException e) {
			if(e.code() != KeeperException.Code.NODEEXISTS)
				throw new RuntimeException(e);

		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public boolean equals(Object o) {
		if (this == o) return true;
		if (o.getClass()!=this.getClass()) return false;

		ZkPrimitive that = (ZkPrimitive) o;

		return baseNode.equals(that.baseNode);
	}

	@Override
	public int hashCode() {
		return baseNode.hashCode();
	}

	protected void setConnectionListener(){
		zkSessionManager.addConnectionListener(connectionListener);
	}

	protected void removeConnectionListener(){
		zkSessionManager.removeConnectionListener(connectionListener);
	}

	protected void notifyParties(){
		localLock.lock();
		try{
			condition.signalAll();
		}finally{
			localLock.unlock();
		}
	}

	private static class PrimitiveConnectionListener extends ConnectionListenerSkeleton{
		private final ZkPrimitive primitive;

		private PrimitiveConnectionListener(ZkPrimitive primitive) {
			this.primitive = primitive;
		}

		@Override
		public void syncConnected() {
			primitive.notifyParties();
		}

		@Override
		public void expired() {
			primitive.broken = true;
			primitive.notifyParties();
		}
	}

	private static class SignallingWatcher implements Watcher{
		private final ZkPrimitive primitive;

		private SignallingWatcher(ZkPrimitive primitive) {
			this.primitive = primitive;
		}

		@Override
		public void process(WatchedEvent event) {

			primitive.notifyParties();
		}
	}
}
